/**
 * 
 */
package com.swg.cocobox.message;

import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.swg.cocobox.KeyChange;
import com.swg.cocobox.message.event.MessageEvent;
import com.swg.cocobox.message.event.MessageEventListener;
import com.swg.cocobox.message.event.MessageEventProcessor;
import com.swg.cocobox.message.event.MessageEventSupport;
import com.swg.cocobox.server.ReadWriteServer;

/**
 * @author satriaprayoga
 *
 */
public abstract class MessageServer extends ReadWriteServer {
	
	private Map<SocketChannel, List<ByteBuffer>> privateMap;
	private Map<SocketChannel, List<ByteBuffer>> broadcastMap;
	
	private MessageEventSupport eventSupport;
	private MessageEventProcessor eventProcessor;
	
	private ClientInfoList clientInfoList;

	private ExecutorService service=Executors.newCachedThreadPool();
	
	public MessageServer(InetAddress address, int port) throws IOException {
		super(address, port);
		this.clientInfoList=new ClientInfoList();
		privateMap=new HashMap<SocketChannel, List<ByteBuffer>>();
		broadcastMap=new HashMap<SocketChannel, List<ByteBuffer>>();
	}
	
	@Override
	public synchronized void startServer() {
		Thread serverThread=new Thread(this);
		serverThread.start();
		service.execute(eventProcessor);
	}
	
	public void sendBroadcast(SocketChannel channel,byte[] data)throws IOException{
		synchronized (keyChanges) {
			keyChanges.add(new KeyChange(channel, KeyChange.CHANGE, SelectionKey.OP_WRITE));
			synchronized (broadcastMap) {
				List<ByteBuffer> queue=broadcastMap.get(channel);
				if(queue==null){
					queue=new ArrayList<ByteBuffer>();
					broadcastMap.put(channel, queue);
				}
				queue.add(ByteBuffer.wrap(data));
			}
		}
		selector.wakeup();
	}
	
	public void sendPrivate(ClientInfo info,byte[] data)throws IOException{
		synchronized (keyChanges) {
			keyChanges.add(new KeyChange(info.getChannel(), KeyChange.CHANGE, SelectionKey.OP_WRITE));
			synchronized (privateMap) {
				List<ByteBuffer> queue=privateMap.get(info.getChannel());
				if(queue==null){
					queue=new ArrayList<ByteBuffer>();
					privateMap.put(info.getChannel(), queue);
				}
				queue.add(ByteBuffer.wrap(data));
			}

		}
		selector.wakeup();
	}
	
	@Override
	protected void accept(SelectionKey key) throws IOException {
		ServerSocketChannel serverSocketChannel=(ServerSocketChannel)key.channel();
		SocketChannel socketChannel=serverSocketChannel.accept();
		socketChannel.configureBlocking(false);
		socketChannel.register(selector, SelectionKey.OP_READ);
		MessageEvent event=eventSupport.fireEvent(socketChannel, MessageContext.CONNECTION_EVENT);
		eventProcessor.queueEvent(event);
	}
	
	@Override
	protected void write(SelectionKey key) throws IOException {
		super.write(key);
		writePrivate(key);
		writeBroadcast(key);
	}

	@Override
	protected void read(SelectionKey key) throws IOException {
		SocketChannel socketChannel=(SocketChannel)key.channel();
		readBuffer.clear();
		int numRead;
		try{
			numRead=socketChannel.read(readBuffer);
		}catch(IOException e){
			key.cancel();
			socketChannel.close();
			return ;
		}
		if(numRead==-1){
			key.cancel();
			key.channel().close();
			return ;
		}
		byte[] copy=new byte[numRead];
		System.arraycopy(readBuffer.array(), 0, copy, 0, numRead);
		MessagePackage messagePackage=new MessagePackage(copy);
		handlePackage(socketChannel,messagePackage);
	}
	
	protected void writeBroadcast(SelectionKey key)throws IOException{
			synchronized (broadcastMap) {
				Set<SocketChannel> channels=broadcastMap.keySet();
				for(SocketChannel socketChannel:channels){
					List<ByteBuffer> queue=broadcastMap.get(socketChannel);
					//	System.out.println(queue==null);
					while(!queue.isEmpty()){
						ByteBuffer buffer=queue.get(0);
						socketChannel.write(buffer);
						if(buffer.remaining()>0)
							break;
						queue.remove(0);
					}
					if(queue.isEmpty()){
						key.interestOps(SelectionKey.OP_READ);
					}
				}
			}
	}
	
	protected void writePrivate(SelectionKey key)throws IOException{
		synchronized (privateMap) {
			Set<SocketChannel> channels=privateMap.keySet();
			for(SocketChannel socketChannel:channels){
				List<ByteBuffer> queue=privateMap.get(socketChannel);
				//	System.out.println(queue==null);
				while(!queue.isEmpty()){
					ByteBuffer buffer=queue.get(0);
					socketChannel.write(buffer);
					if(buffer.remaining()>0)
						break;
					queue.remove(0);
				}
				if(queue.isEmpty()){
					key.interestOps(SelectionKey.OP_READ);
				}
			}
		}
	}
	
	protected abstract void handlePackage(SocketChannel channel,MessagePackage messagePackage);
	
	public void setEventProcessor(MessageEventProcessor eventProcessor) {
		this.eventProcessor = eventProcessor;
	}
	
	public void setEventSupport(MessageEventSupport eventSupport) {
		this.eventSupport = eventSupport;
	}
	
	public MessageEventProcessor getEventProcessor() {
		return eventProcessor;
	}
	
	public MessageEventSupport getEventSupport() {
		return eventSupport;
	}
	
	public void addListener(MessageEventListener listener){
		this.eventSupport.addListener(listener);
	}
	
	public void fireConnectionEvent(MessageEvent event){
		if(!eventSupport.getListeners().isEmpty()){
			for(MessageEventListener listener:eventSupport.getListeners()){
				listener.onConnection(event);
			}
		}
	}
	
	public void fireReceivingEvent(MessageEvent event){
		if(!eventSupport.getListeners().isEmpty()){
			for(MessageEventListener listener:eventSupport.getListeners()){
				listener.onReceiving(event);
			}
		}
	}
	
	public ClientInfoList getClientInfoList() {
		return clientInfoList;
	}

}
